package com.example.mdc.logback.module.service;

import com.alibaba.google.common.collect.Maps;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.example.mdc.logback.common.config.canal.CanalConfig;
import com.example.mdc.logback.common.config.mq.RocketMQProducer;
import com.example.mdc.logback.common.utils.RedisUtils;
import com.example.mdc.logback.module.enitity.Balance;
import jakarta.annotation.PostConstruct;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import top.javatool.canal.client.handler.EntryHandler;

import java.util.List;
import java.util.Map;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;

/**
 * @author: liu_pc
 * Date:        2023/8/25
 * Description: 余额信息变更Redis变成处理类
 * Version:     1.0
 */
@Component(value = "BalanceRedisProcessor")
public class BalanceRedisProcessorService implements EntryHandler<Balance>, Runnable {

    private final Logger logger = LoggerFactory.getLogger(BalanceRedisProcessorService.class);

    private final RedisUtils redisUtils;

    private final CanalConfig canalConfig;

    private final Executor executor;

    private final RocketMQProducer rocketMQProducer;

    @Value("${canal.server.open}")
    private boolean open;

    /**
     * 重试次数
     */
    private final static int MAX_RETRY_COUNT = 3;

    @Autowired(required = true)
    public BalanceRedisProcessorService(RedisUtils redisUtils,
                                        CanalConfig canalConfig,
                                        @Qualifier("ownThreadPoolExecutor") Executor executor, RocketMQProducer rocketMqProducer) {
        this.redisUtils = redisUtils;
        this.canalConfig = canalConfig;
        this.executor = executor;
        this.rocketMQProducer = rocketMqProducer;
    }


    @PostConstruct
    public void init() {
        Map<String, String> mainMdcContext = Maps.newHashMap();
        mainMdcContext.put("canal-thread", "balance-redis-processor-service");
        MDC.setContextMap(mainMdcContext);
        executor.execute(this);
        logger.info("MySQL-Balance数据自动同步到Redis：线程已经启动");
    }

    @Override
    public void run() {
        CanalConnector canalConnector = canalConfig.canalConnector();
        canalConnector.connect();
        // 回滚到未进行ack的地方
        canalConnector.rollback();
        try {
            while (open) {
                // 获取数据 每次获取一百条改变数据
                Message message = canalConnector.getWithoutAck(100);
                //获取这条消息的id
                long batchId = message.getId();
                int size = message.getEntries().size();

                if (batchId == -1 || size == 0) {
                    TimeUnit.SECONDS.sleep(1);
                    continue;
                }

                // 处理数据
                for (CanalEntry.Entry entry : message.getEntries()) {
                    if (entry.getEntryType() != CanalEntry.EntryType.ROWDATA) {
                        continue;
                    }

                    CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                    CanalEntry.EventType eventType = rowChange.getEventType();
                    List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();

                    boolean syncRedisDataFlag = eventType == CanalEntry.EventType.UPDATE
                            || eventType == CanalEntry.EventType.INSERT
                            || eventType == CanalEntry.EventType.DELETE;
                    if (!syncRedisDataFlag) {
                        continue;
                    }

                    for (CanalEntry.RowData rowData : rowDatasList) {
                        List<CanalEntry.Column> columns = rowData.getAfterColumnsList();
                        String tableName = entry.getHeader().getTableName();

                        // 判断是否是 Balance 表的 amount 字段变更
                        if (!"balance".equals(tableName)) {
                            return;
                        }

                        StringBuilder redisKey = new StringBuilder("balance:");
                        handleCanalChangeColumns(columns, redisKey);
                    }
                }
                // 确认消费完成这条消息
                canalConnector.ack(message.getId());
            }
        } catch (Exception e) {
            logger.error("canal-数据同步异常");
            //运行时异常，服务监控告警，需要开发介入排查
            throw new RuntimeException(e);
        } finally {
            // 关闭连接
            canalConnector.disconnect();
        }
    }


    /**
     * 开始处理canal获取到的变更列到Redis
     *
     * @param columns  列
     * @param redisKey Redis中数据存储的Key
     * @throws InterruptedException 异常
     */
    private void handleCanalChangeColumns(List<CanalEntry.Column> columns, StringBuilder redisKey) throws Exception {
        String changeInfo = null;
        for (CanalEntry.Column column : columns) {
            logger.info("Balance changed in 'balance' dataInfo: {}", column);
            if ("id".equals(column.getName())) {
                String changeId = column.getValue();
                logger.info("当前变更id为:{}", changeId);
                redisKey.append(changeId);
            }
            if ("amount".equals(column.getName())) {
                String changeValue = column.getValue();
                boolean success = false;
                logger.info(changeValue);
                for (int retryCount = 0; retryCount < MAX_RETRY_COUNT; retryCount++) {
                    try {
                        redisUtils.set(redisKey.toString(), changeValue);
                        success = true;
                        logger.info("消费成功");
                        return;
                    } catch (Exception ex) {
                        logger.error("存入Redis失败，进行重试：{}", ex.getMessage());
                        // 等待一段时间后进行重试
                        TimeUnit.SECONDS.sleep(1);
                    }
                    changeInfo = redisKey.append(":").append(changeValue).toString();
                }

                //发送告警消息
                if (!success) {
                    rocketMQProducer.sendMessage("DefaultCluster", changeInfo);
                }
            }
        }
    }
}
